/*
 * Copyright 2021-2022 Open Kunlun Technology <https://www.openkunlun.io>
 */

package io.openkunlun.scaladsl.serialization

import akka.actor.{ ExtendedActorSystem, Extension, ExtensionId, ExtensionIdProvider }
import akka.serialization.{ SerializationExtension, SerializerWithStringManifest }
import com.google.protobuf.ByteString

/**
 * @author ericxin.
 */
object AkkaDelegateSerialization extends ExtensionId[AkkaDelegateSerialization] with ExtensionIdProvider {
  override def lookup: ExtensionId[AkkaDelegateSerialization] = AkkaDelegateSerialization
  override def createExtension(system: ExtendedActorSystem) = new AkkaDelegateSerialization(system)
}

class AkkaDelegateSerialization(system: ExtendedActorSystem) extends Extension with DaprObjectSerialization {
  val serialization = SerializationExtension(system)

  val transform: DelegatingTransformer.Transformer = DelegatingTransformer.empty()

  /**
   *
   * @param serializable
   * @return
   */
  override def serialize[T](serializable: T): Array[Byte] = {
    val o = serializable.asInstanceOf[AnyRef]
    val serializer = serialization.findSerializerFor(o)
    val (manifest, isStringManifest) = serializer match {
      case ser: SerializerWithStringManifest => (ser.manifest(o), true)
      case _                                 => (if (serializer.includeManifest) serializable.getClass.getName else "", false)
    }
    DelegateManifest(
      if (transform.isDefinedAt(manifest)) transform(manifest) else manifest,
      ByteString.copyFrom(serializer.toBinary(o)),
      isStringManifest,
      serializer.identifier).toByteArray
  }

  /**
   *
   * @param bytes
   * @tparam T
   * @return
   */
  override def deserialize[T: Manifest](bytes: Array[Byte]): T = {
    val format = DelegateManifest.parseFrom(bytes)
    val manifest = if (transform.isDefinedAt(format.manifest)) transform(format.manifest) else format.manifest
    if (format.isStringManifest) {
      serialization.deserialize(format.payload.toByteArray, format.serializerId, manifest).get.asInstanceOf[T]
    } else if (format.manifest.nonEmpty) {
      val clazz = system.dynamicAccess.getClassFor[T](manifest).get
      serialization.deserialize(format.payload.toByteArray, clazz).get
    } else {
      serialization.deserialize[T](format.payload.toByteArray, format.serializerId, None).get
    }
  }
}
